package com.kexilo.messaging.autoconfigure;

import com.kexilo.messaging.properties.MessageProperties;
import com.kexilo.messaging.service.MessageService;
import com.kexilo.messaging.publisher.MessagePublisher;
import com.kexilo.messaging.processor.MessageProcessor;
import com.kexilo.messaging.domain.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.*;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.Map;
import java.util.List;
import java.util.ArrayList;

/**
 * 消息处理插件自动配置类
 * 根据条件自动装配消息处理相关组件
 * 
 * @author Kexilo
 */
@AutoConfiguration
@ConditionalOnClass({MessageService.class})
@ConditionalOnProperty(value = "kexilo.plugin.messaging.enabled", havingValue = "true", matchIfMissing = true)
@EnableConfigurationProperties(MessageProperties.class)
@EnableAsync
public class MessageAutoConfiguration {
    
    private static final Logger log = LoggerFactory.getLogger(MessageAutoConfiguration.class);
    
    /**
     * 配置消息服务
     */
    @Bean
    @ConditionalOnMissingBean(MessageService.class)
    public MessageService messageService(MessageProperties properties) {
        log.info("配置消息服务，中间件类型: {}", properties.getMiddlewareType());
        return new SimpleMessageService(properties);
    }
    
    /**
     * 配置消息发布者
     */
    @Bean
    @ConditionalOnMissingBean(MessagePublisher.class)
    public MessagePublisher messagePublisher(MessageService messageService, MessageProperties properties) {
        log.info("配置消息发布者");
        return new SimpleMessagePublisher(messageService, properties);
    }
    
    /**
     * 配置消息处理器注册表
     */
    @Bean
    @ConditionalOnMissingBean(MessageProcessorRegistry.class)
    public MessageProcessorRegistry messageProcessorRegistry(MessageProperties properties) {
        log.info("配置消息处理器注册表");
        return new MessageProcessorRegistry(properties);
    }
    
    /**
     * 配置消息任务执行器
     */
    @Bean("messagingTaskExecutor")
    @ConditionalOnMissingBean(name = "messagingTaskExecutor")
    public ThreadPoolTaskExecutor messagingTaskExecutor(MessageProperties properties) {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(properties.getProducer().getThreadPoolSize());
        executor.setMaxPoolSize(properties.getProducer().getThreadPoolSize() * 2);
        executor.setQueueCapacity(500);
        executor.setThreadNamePrefix("kexilo-messaging-");
        executor.setKeepAliveSeconds(60);
        executor.setAllowCoreThreadTimeOut(false);
        
        // 拒绝策略：调用者运行
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        
        // 等待任务完成后关闭线程池
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.setAwaitTerminationSeconds(60);
        
        executor.initialize();
        
        log.info("配置消息异步执行器，线程池大小: {}", properties.getProducer().getThreadPoolSize());
        return executor;
    }
    
    /**
     * 消息监控调度器
     */
    @Bean
    @ConditionalOnProperty(value = "kexilo.plugin.messaging.monitor.enabled", havingValue = "true", matchIfMissing = true)
    public MessageMonitorScheduler messageMonitorScheduler(MessageService messageService, MessageProperties properties) {
        log.info("配置消息监控调度器");
        return new MessageMonitorScheduler(messageService, properties);
    }
    
    /**
     * 死信队列清理调度器
     */
    @Bean
    @ConditionalOnProperty(value = "kexilo.plugin.messaging.dead-letter-queue-enabled", havingValue = "true", matchIfMissing = true)
    public DeadLetterCleanupScheduler deadLetterCleanupScheduler(MessageService messageService, MessageProperties properties) {
        log.info("配置死信队列清理调度器");
        return new DeadLetterCleanupScheduler(messageService, properties);
    }
    
    /**
     * 消息健康检查
     */
    @Bean
    @ConditionalOnClass(name = "org.springframework.boot.actuate.health.HealthIndicator")
    @ConditionalOnProperty(value = "kexilo.plugin.messaging.health.enabled", havingValue = "true")
    public MessageHealthIndicator messageHealthIndicator(MessageService messageService, MessageProperties properties) {
        log.info("配置消息健康检查");
        return new MessageHealthIndicator(messageService, properties);
    }
    
    /**
     * 初始化消息主题
     */
    @Bean
    @ConditionalOnProperty(value = "kexilo.plugin.messaging.auto-create-topics", havingValue = "true")
    public ApplicationRunner messageTopicInitializer(MessageService messageService, MessageProperties properties) {
        return args -> {
            log.info("开始初始化消息主题...");
            try {
                for (Map.Entry<String, MessageProperties.TopicProperties> entry : properties.getTopics().entrySet()) {
                    String topicName = entry.getKey();
                    MessageProperties.TopicProperties topicProps = entry.getValue();
                    messageService.createTopic(topicName, topicProps.getPartitions(), topicProps.getReplicas());
                    log.info("创建主题: {}, 分区数: {}, 副本数: {}", topicName, topicProps.getPartitions(), topicProps.getReplicas());
                }
                log.info("消息主题初始化完成");
            } catch (Exception e) {
                log.warn("消息主题初始化失败", e);
            }
        };
    }
    
    /**
     * 简化的消息服务实现
     */
    public static class SimpleMessageService implements MessageService {
        
        private final MessageProperties properties;
        private final Map<String, Message> messageStore = new ConcurrentHashMap<>();
        private final Map<String, MessageProcessor> processors = new ConcurrentHashMap<>();
        private final Logger log = LoggerFactory.getLogger(SimpleMessageService.class);
        
        public SimpleMessageService(MessageProperties properties) {
            this.properties = properties;
        }
        
        @Override
        public boolean sendMessage(Message message) {
            try {
                message.setStatus(Message.MessageStatus.SENT);
                message.setSendTime(java.time.LocalDateTime.now());
                messageStore.put(message.getMessageId(), message);
                log.debug("发送消息成功: {}", message.getMessageId());
                return true;
            } catch (Exception e) {
                log.error("发送消息失败: {}", message.getMessageId(), e);
                return false;
            }
        }
        
        @Override
        public CompletableFuture<Boolean> sendMessageAsync(Message message) {
            return CompletableFuture.supplyAsync(() -> sendMessage(message));
        }
        
        @Override
        public int batchSendMessages(List<Message> messages) {
            int successCount = 0;
            for (Message message : messages) {
                if (sendMessage(message)) {
                    successCount++;
                }
            }
            return successCount;
        }
        
        @Override
        public boolean sendDelayMessage(Message message, long delayMillis) {
            message.setDelayTime(delayMillis);
            return sendMessage(message);
        }
        
        @Override
        public boolean sendScheduledMessage(Message message, long timestamp) {
            long delayMillis = timestamp - System.currentTimeMillis();
            return sendDelayMessage(message, Math.max(0, delayMillis));
        }
        
        @Override
        public boolean subscribe(String topic, MessageProcessor processor) {
            processors.put(topic, processor);
            log.info("订阅主题: {}", topic);
            return true;
        }
        
        @Override
        public boolean unsubscribe(String topic) {
            processors.remove(topic);
            log.info("取消订阅主题: {}", topic);
            return true;
        }
        
        @Override
        public Message getMessageById(String messageId) {
            return messageStore.get(messageId);
        }
        
        @Override
        public boolean isAvailable() {
            return true;
        }
        
        @Override
        public Map<String, Object> getHealthStatus() {
            Map<String, Object> status = new java.util.HashMap<>();
            status.put("available", true);
            status.put("middlewareType", properties.getMiddlewareType());
            status.put("messageCount", messageStore.size());
            status.put("subscriberCount", processors.size());
            return status;
        }
        
        // 其他方法的简化实现...
        @Override public List<Message> pullMessages(String topic, int maxMessages) { return java.util.Collections.emptyList(); }
        @Override public boolean acknowledgeMessage(String messageId) { return true; }
        @Override public boolean rejectMessage(String messageId, boolean requeue) { return true; }
        @Override public List<Message> getMessagesByBusinessId(String businessId) { return java.util.Collections.emptyList(); }
        @Override public List<Message> getMessagesByStatus(Message.MessageStatus status, int limit) { return java.util.Collections.emptyList(); }
        @Override public boolean retryMessage(String messageId) { return true; }
        @Override public int batchRetryMessages(List<String> messageIds) { return 0; }
        @Override public boolean deleteMessage(String messageId) { return messageStore.remove(messageId) != null; }
        @Override public int cleanupExpiredMessages(int daysBeforeNow) { return 0; }
        @Override public List<Message> getDeadLetterMessages(String topic, int limit) { return java.util.Collections.emptyList(); }
        @Override public boolean handleDeadLetterMessage(String messageId, String action) { return true; }
        @Override public Map<String, Object> getMessageStatistics() { return Map.of("totalMessages", messageStore.size()); }
        @Override public Map<String, Object> getTopicStatistics(String topic) { return Map.of("topic", topic); }
        @Override public List<Map<String, Object>> getMessageTrends(String topic, int hours) { return java.util.Collections.emptyList(); }
        @Override public boolean pauseConsumer(String topic) { return true; }
        @Override public boolean resumeConsumer(String topic) { return true; }
        @Override public String getConsumerStatus(String topic) { return "RUNNING"; }
        @Override public boolean createTopic(String topic, int partitions, int replicas) { return true; }
        @Override public boolean deleteTopic(String topic) { return true; }
        @Override public List<String> getAllTopics() { return new ArrayList<>(processors.keySet()); }
        @Override public String exportMessages(String topic, String startTime, String endTime, String format) { return "/tmp/messages.json"; }
    }
    
    /**
     * 简化的消息发布者实现
     */
    public static class SimpleMessagePublisher implements MessagePublisher {
        
        private final MessageService messageService;
        private final MessageProperties properties;
        private final Logger log = LoggerFactory.getLogger(SimpleMessagePublisher.class);
        
        public SimpleMessagePublisher(MessageService messageService, MessageProperties properties) {
            this.messageService = messageService;
            this.properties = properties;
        }
        
        @Override
        public PublishResult publish(String topic, String payload) {
            try {
                Message message = Message.builder()
                    .topic(topic)
                    .payload(payload)
                    .build();
                
                boolean success = messageService.sendMessage(message);
                return success ? PublishResult.success(message.getMessageId()) 
                              : PublishResult.failure("发布失败");
            } catch (Exception e) {
                log.error("发布消息失败: topic={}", topic, e);
                return PublishResult.failure("发布异常: " + e.getMessage(), e);
            }
        }
        
        @Override
        public PublishResult publish(Message message) {
            try {
                boolean success = messageService.sendMessage(message);
                return success ? PublishResult.success(message.getMessageId()) 
                              : PublishResult.failure("发布失败");
            } catch (Exception e) {
                log.error("发布消息失败: messageId={}", message.getMessageId(), e);
                return PublishResult.failure("发布异常: " + e.getMessage(), e);
            }
        }
        
        @Override
        public CompletableFuture<PublishResult> publishAsync(String topic, String payload) {
            return CompletableFuture.supplyAsync(() -> publish(topic, payload));
        }
        
        @Override
        public CompletableFuture<PublishResult> publishAsync(Message message) {
            return CompletableFuture.supplyAsync(() -> publish(message));
        }
        
        @Override
        public boolean isAvailable() {
            return messageService.isAvailable();
        }
        
        @Override
        public Map<String, Object> getHealthStatus() {
            return messageService.getHealthStatus();
        }
        
        // 其他方法的简化实现...
        @Override public BatchPublishResult batchPublish(List<Message> messages) { return null; }
        @Override public PublishResult publishWithTag(String topic, String tag, String payload) { return publish(topic, payload); }
        @Override public PublishResult publishWithPriority(String topic, String payload, int priority) { return publish(topic, payload); }
        @Override public PublishResult publishDelayMessage(String topic, String payload, long delayMillis) { return publish(topic, payload); }
        @Override public PublishResult publishScheduledMessage(String topic, String payload, long timestamp) { return publish(topic, payload); }
        @Override public PublishResult publishTransactionalMessage(String topic, String payload, String transactionId) { return publish(topic, payload); }
        @Override public boolean commitTransaction(String transactionId) { return true; }
        @Override public boolean rollbackTransaction(String transactionId) { return true; }
        @Override public PublishResult publishBroadcast(String topic, String payload) { return publish(topic, payload); }
        @Override public PublishResult publishPointToPoint(String queue, String payload) { return publish(queue, payload); }
        @Override public CompletableFuture<String> publishRpcRequest(String topic, String payload, long timeout) { return CompletableFuture.completedFuture("response"); }
        @Override public PublishResult publishNotification(String notificationType, List<String> recipients, String title, String content) { return PublishResult.success("notification-sent"); }
        @Override public PublishResult publishEmail(String to, String subject, String content, List<String> attachments) { return PublishResult.success("email-sent"); }
        @Override public PublishResult publishSms(String phoneNumber, String content, String template) { return PublishResult.success("sms-sent"); }
        @Override public PublishResult publishPush(String deviceId, String title, String content, Map<String, Object> extra) { return PublishResult.success("push-sent"); }
        @Override public Map<String, Object> getPublishStatistics() { return Map.of("totalPublished", 0); }
        @Override public Map<String, Object> getTopicPublishStatistics(String topic) { return Map.of("topic", topic); }
    }
    
    /**
     * 消息处理器注册表
     */
    public static class MessageProcessorRegistry {
        
        private final Map<String, MessageProcessor> processors = new ConcurrentHashMap<>();
        private final MessageProperties properties;
        private final Logger log = LoggerFactory.getLogger(MessageProcessorRegistry.class);
        
        public MessageProcessorRegistry(MessageProperties properties) {
            this.properties = properties;
        }
        
        public void registerProcessor(String processorName, MessageProcessor processor) {
            processors.put(processorName, processor);
            log.info("注册消息处理器: {}", processorName);
        }
        
        public void unregisterProcessor(String processorName) {
            processors.remove(processorName);
            log.info("注销消息处理器: {}", processorName);
        }
        
        public MessageProcessor getProcessor(String processorName) {
            return processors.get(processorName);
        }
        
        public Map<String, MessageProcessor> getAllProcessors() {
            return new java.util.HashMap<>(processors);
        }
    }
    
    /**
     * 消息监控调度器
     */
    public static class MessageMonitorScheduler {
        
        private final MessageService messageService;
        private final MessageProperties properties;
        private final Logger log = LoggerFactory.getLogger(MessageMonitorScheduler.class);
        
        public MessageMonitorScheduler(MessageService messageService, MessageProperties properties) {
            this.messageService = messageService;
            this.properties = properties;
        }
        
        @Scheduled(fixedDelayString = "#{@messageProperties.monitor.collectInterval.toMillis()}")
        public void collectMetrics() {
            if (properties.getMonitor().isEnabled()) {
                try {
                    Map<String, Object> stats = messageService.getMessageStatistics();
                    log.debug("消息统计: {}", stats);
                } catch (Exception e) {
                    log.error("收集消息监控数据失败", e);
                }
            }
        }
    }
    
    /**
     * 死信队列清理调度器
     */
    public static class DeadLetterCleanupScheduler {
        
        private final MessageService messageService;
        private final MessageProperties properties;
        private final Logger log = LoggerFactory.getLogger(DeadLetterCleanupScheduler.class);
        
        public DeadLetterCleanupScheduler(MessageService messageService, MessageProperties properties) {
            this.messageService = messageService;
            this.properties = properties;
        }
        
        @Scheduled(fixedDelayString = "#{@messageProperties.deadLetter.cleanupInterval.toMillis()}")
        public void cleanupDeadLetters() {
            if (properties.getDeadLetter().isAutoCleanupEnabled()) {
                try {
                    log.info("开始清理过期死信消息");
                    // 简化实现
                    log.info("死信消息清理完成");
                } catch (Exception e) {
                    log.error("清理死信消息失败", e);
                }
            }
        }
    }
    
    /**
     * 消息健康检查指示器
     */
    public static class MessageHealthIndicator {
        
        private final MessageService messageService;
        private final MessageProperties properties;
        
        public MessageHealthIndicator(MessageService messageService, MessageProperties properties) {
            this.messageService = messageService;
            this.properties = properties;
        }
        
        public String checkHealth() {
            try {
                if (!properties.isEnabled()) {
                    return "DISABLED";
                }
                
                boolean available = messageService.isAvailable();
                return available ? "UP" : "DOWN";
                
            } catch (Exception e) {
                return "DOWN: " + e.getMessage();
            }
        }
        
        public String getServiceInfo() {
            return String.format("MessageService[%s] - MiddlewareType: %s, SerializationType: %s", 
                    messageService.getServiceName(),
                    properties.getMiddlewareType(),
                    properties.getSerializationType());
        }
        
        public Map<String, Object> getDetailedHealth() {
            return messageService.getHealthStatus();
        }
    }
}
